package com.hph.springcloudstreamkafkademo.stream.consumer;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * 消息消费bean（spring-cloud-stream-kafka方式）
 * @author HPH
 * @date 2019/7/27
 */
@Component
@EnableBinding({Sink.class})
public class MessageConsumerBean {

    private final SubscribableChannel subscribableChannel;

    public MessageConsumerBean(@Qualifier(Sink.INPUT) SubscribableChannel subscribableChannel) {
        this.subscribableChannel = subscribableChannel;
    }

    /**
     * 方式1：与第二、三种方式互斥（即第二、三种不会打印）
     * 当字段注入完成后的回调
     */
    @PostConstruct
    public void init(){

        //实现异步回调
        subscribableChannel.subscribe(message -> System.out.println("消息消费者第一种方式：" + message.getPayload()));
    }

    /**
     * 方式二：与第一、三种方式互斥（即第一、三种不会打印）
     * 通过@ServiceActivator实现
     * @param message 消息对象
     */
    @ServiceActivator(inputChannel = Sink.INPUT)
    public void onMessage(Object message){

        System.out.println("消息消费者第二种方式：" + message);
    }

    /**
     * 方式三：与第一、二种方式互斥（即第一、二种不会打印）
     * 通过@StreamListener实现
     * @param message 消息
     */
    @StreamListener(Sink.INPUT)
    public void onMessage(String message){

        System.out.println("消息消费者第三种方式：" + message);
    }
}
